package com.lmax.disruptor.custom.java8;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class LongEventMainJava8
{
    /**
     * 用lambda表达式来注册EventHandler和EventProductor
     *
     * @param args
     * @throws InterruptedException
     */
    public static void main(final String[] args) throws InterruptedException
    {
        // Executor that will be used to construct new threads for consumers
        Executor executor = Executors.newCachedThreadPool();
        // Specify the size of the ring buffer, must be power of 2.
        int bufferSize = 1024;
        // Construct the Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);
        // 可以使用lambda来注册一个EventHandler
        EventHandler<LongEvent> eventHandler = (event, sequence, endOfBatch) ->
        {
            TimeUnit.SECONDS.sleep(1);
            System.out.println("sequence===" + sequence + "  Event: " + event.getValue());
        };
        disruptor.handleEventsWith(eventHandler);
        // Start the Disruptor, starts all threads running
        disruptor.start();
        // Get the ring buffer from the Disruptor to be used for publishing.
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

        LongEventProducer producer = new LongEventProducer(ringBuffer);

        AtomicLong atomicLong = new AtomicLong(0L);
        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long l = 0; l < 1_000_000; l++)
        {
            bb.putLong(0, l);
            EventTranslatorOneArg<LongEvent, ByteBuffer> eventTranslatorOneArg = (event, sequence, buffer)
                    -> event.setValue(buffer.getLong(0));
            // 填充事件的业务数据
            System.out.println("producer ---" + atomicLong.getAndIncrement());
            ringBuffer.publishEvent(eventTranslatorOneArg, bb);
//TimeUnit.NANOSECONDS.sleep(5);
        }
    }
}